package com.apps.sdses.flink141.demo;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;

/**
 * @author jiwei
 * @description
 * @date 2023/9/15 15:38
 */
public class HeatPipelineLeakageDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<Tuple3<String, Integer, Long>> dataStream = env.fromElements(
                Tuple3.of("SensorA", 100, 1694742900000L),   //2023-09-15 09:55:00
                Tuple3.of("SensorA", 101, 1694743200000L),   //2023-09-15 10:00:00
                Tuple3.of("SensorA", 102, 1694743202000L),   //2023-09-15 10:00:02
                Tuple3.of("SensorA", 103, 1694743204000L),   //2023-09-15 10:00:04
                Tuple3.of("SensorA", 104, 1694743206000L),   //2023-09-15 10:00:06
                Tuple3.of("SensorA", 105, 1694743208000L),   //2023-09-15 10:00:08
                Tuple3.of("SensorA", 106, 1694743210000L),   //2023-09-15 10:00:10
                Tuple3.of("SensorA", 107, 1694743212000L),   //2023-09-15 10:00:12

                Tuple3.of("SensorB", 91, 1694743140000L),   //2023-09-15 09:59:00
                Tuple3.of("SensorB", 92, 1694743200000L),   //2023-09-15 10:00:00
                Tuple3.of("SensorB", 93, 1694743202000L),   //2023-09-15 10:00:02
                Tuple3.of("SensorB", 4, 1694743204000L),   //2023-09-15 10:00:04
                Tuple3.of("SensorB", 5, 1694743206000L),   //2023-09-15 10:00:06
                Tuple3.of("SensorB", 6, 1694743208000L),   //2023-09-15 10:00:08
                Tuple3.of("SensorB", 7, 1694743215000L)    //2023-09-15 10:00:15
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, Integer, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((event, timestamp) -> event.f2));

        //不分组后开窗
        dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, Long>> iterable, Collector<String> collector) throws Exception {
                        SimpleDateFormat outputFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String start = outputFormat.format(new Date(context.window().getStart()));
                        String end = outputFormat.format(new Date(context.window().getEnd()));
                        HashMap<Tuple3<String, Integer, Long>, Tuple3<String, Integer, Long>> map = new HashMap<>();

                        LinkedList<Tuple3<String, Integer, Long>> eventA = new LinkedList<>();
                        LinkedList<Tuple3<String, Integer, Long>> eventB = new LinkedList<>();

                        for (Tuple3<String, Integer, Long> event : iterable) {
                            if (event.f0.equals("SensorA")) {
                                eventA.add(event);
                            } else if (event.f0.equals("SensorB")) {
                                eventB.add(event);
                            } else {
                                System.out.println("脏数据: " + event + ",被过滤掉...");
                            }
                        }
                        //TODO 查找SensorA时间最近的SensorB
                        for (Tuple3<String, Integer, Long> a : eventA) {
                            map.put(a, findClosestB(a, eventB));
                        }

                        //TODO 遍历A和最近的B, 判断是否满足预警规则
                        for (Tuple3<String, Integer, Long> event : map.keySet()) {
                            double vca = Double.parseDouble(event.f1.toString());
                            Tuple3<String, Integer, Long> eb = map.get(event);
                            if (eb != null) {
                                double vcb = Double.parseDouble(eb.f1.toString());

                                double v = (vca - vcb) / vca;
                                System.out.println("-------------差值： " + v);
                                if (v > 0.1) {
                                    collector.collect("发生泄露!!!, SensorA: " + event + ", SensorB: " + eb);
                                } else {
                                    collector.collect("没有发生泄露. SensorA: " + event + ", SensorB: " + eb);
                                }
                            }
                        }
                    }
                })
                .print();

        env.execute();
    }

    private static Tuple3<String, Integer, Long> findClosestB(Tuple3<String, Integer, Long> eventA, List<Tuple3<String, Integer, Long>> eventBs) {
        Long aTs = eventA.f2;
        long closestDiff = Long.MAX_VALUE;
        Tuple3<String, Integer, Long> b = null;
        for (Tuple3<String, Integer, Long> eventB : eventBs) {
            long diff = Math.abs(eventB.f2 - aTs);
            if (diff < closestDiff) {
                b = eventB;
                closestDiff = diff;
            }
        }
        return b;

    }
}
